package com.bw.gmall.realtime.utils;

import com.bw.gmall.realtime.bean.TransientSink;
import com.bw.gmall.realtime.common.GmallConfig;
import lombok.SneakyThrows;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Field;
import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.util.Date;

/*
*
* ClickHouse写入工具类
*
*
*/
public class MyClickHouseUtil {
    private static final Logger log = LoggerFactory.getLogger(MyClickHouseUtil.class);

    //T  代表输入类型
    public static <T> SinkFunction<T> getSinkFunction(String sql) {
        // 记录SQL和调用类信息，便于调试
        log.info("创建ClickHouse Sink，SQL: {}", sql);
        log.info("调用者: {}", Thread.currentThread().getStackTrace()[2].getClassName());

        return JdbcSink.<T>sink(
                sql,
                new JdbcStatementBuilder<T>() {
                    @SneakyThrows
                    @Override
                    public void accept(PreparedStatement preparedStatement, T t) throws SQLException {
                        log.debug("准备插入数据到ClickHouse: {}", t);
                        try {
                            //使用反射的方式获取t对象中的数据
                            Class<?> tClz = t.getClass();

                            //获取并遍历属性
                            Field[] declaredFields = tClz.getDeclaredFields();

                            int offset = 0;   //用于跳过带TransientSink注解的字段
                            int paramIndex = 0; // 实际参数索引
                            for (int i = 0; i < declaredFields.length; i++) {
                                //获取单个属性
                                Field field = declaredFields[i];
                                //暴力访问
                                field.setAccessible(true);

                                //尝试获取字段上的自定义注解
                                TransientSink transientSink = field.getAnnotation(TransientSink.class);

                                if (transientSink != null) {
                                    log.debug("跳过带TransientSink注解的字段: {}", field.getName());
                                    offset++; //遇到注解，增加偏移量
                                    continue;
                                }

                                paramIndex++;
                                //获取属性值
                                Object value = null;
                                try {
                                    value = field.get(t);
                                } catch (IllegalAccessException e) {
                                    log.error("获取字段[{}]值失败: {}", field.getName(), e.getMessage());
                                    // 如果获取值失败，抛出异常中断处理
                                    throw new SQLException("无法获取字段值: " + field.getName(), e);
                                }

                                //确保非空值
                                if (value == null) {
                                    log.debug("字段[{}]值为null，设置默认值", field.getName());
                                    if (field.getType() == String.class) {
                                        value = "";//空字符串
                                    } else if (field.getType() == Long.class || field.getType() == Integer.class || field.getType() == int.class || field.getType() == long.class) {
                                        value = 0L;//长整型0
                                    } else if (field.getType() == Double.class || field.getType() == double.class) {
                                        value = 0.0D;//双精度0
                                    } else if (field.getType() == BigDecimal.class) {
                                        value = BigDecimal.ZERO;
                                    }
                                }

                                //处理Date类型字段，转换为Timestamp对象以兼容ClickHouse的DateTime类型
                                if (value instanceof Date) {
                                    Date dateValue = (Date) value;
                                    Timestamp timestamp = new Timestamp(dateValue.getTime());
                                    preparedStatement.setTimestamp(paramIndex, timestamp);
                                    log.debug("设置参数[{}]为日期类型值: {}", paramIndex, timestamp);
                                } else {
                                    //给占位符赋值
                                    preparedStatement.setObject(paramIndex, value);
                                    log.debug("设置参数[{}]为: {}", paramIndex, value);
                                }

                            }
                            // 添加日志确认参数设置完成
                            log.debug("成功设置所有参数，准备执行SQL: {}", sql);
                        } catch (Exception e) {
                            log.error("准备ClickHouse插入数据时出错: {}", e.getMessage(), e);
                            // 记录详细的对象信息，方便调试
                            log.error("出错的数据对象: {}", t);
                            // 抛出异常，确保问题能被上层捕获
                            throw new SQLException("ClickHouse插入数据准备失败", e);
                        }
                    }
                },
                new JdbcExecutionOptions.Builder()
                        .withBatchSize(5)
                        .withBatchIntervalMs(1000L)
                        .withMaxRetries(3) // 添加重试机制
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName(GmallConfig.CLICKHOUSE_DRIVER)
                        .withUrl(GmallConfig.CLICKHOUSE_URL)
                        .build());
    }

    /**
     * 用于测试ClickHouse连接是否正常的简单方法
     */
    public static boolean testConnection() {
        Connection conn = null;
        try {
            Class.forName(GmallConfig.CLICKHOUSE_DRIVER);
            conn = DriverManager.getConnection(GmallConfig.CLICKHOUSE_URL);
            log.info("ClickHouse连接测试成功");
            return true;
        } catch (Exception e) {
            log.error("ClickHouse连接测试失败: {}", e.getMessage(), e);
            return false;
        } finally {
            if (conn != null) {
                try {
                    conn.close();
                } catch (SQLException e) {
                    log.error("关闭ClickHouse连接时出错: {}", e.getMessage());
                }
            }
        }
    }
}
